#!/usr/bin/python3

import pymysql
import pymongo
from pymongo import CursorType
import datetime
from bson.timestamp import Timestamp
import decimal


def bson_ts_to_long(timestamp: Timestamp):
    """时间戳转数字"""
    return (timestamp.time << 32) + timestamp.inc


def long_to_bson_ts(val: int):
    """数字转时间戳
    """
    seconds = val >> 32
    increment = val & 0xFFFFFFFF

    return Timestamp(seconds, increment)

class SqlSata:
    def __init__(self, db, tablename):
        self.tablename = tablename
        self.db = db
        self.up_list = []
        self.add_list = []
        self.del_list = []
        self.other_ids = []
        self.other_list = []
        self.count = 0
        self.type = "m"

    def get_val(self, v):
        t = type(v).__name__
        # print(t)
        if t == "UUID":
            return str(v)
        if t == "Int64":
            return int(v)
        if t == "Decimal128":            
            return v.to_decimal()
        return v

    def get(self, data, key, d):
        keys = key.split(".")
        rt = data
        for k in keys:            
            rt = self._get(k, rt)
            if rt == None:
                return d            
        return rt

    def _get(self, k, data):
        if k in data:
            return data[k]
        return None
    
    def convert(self, data):
        return data

    def del_data(self, ids):
        sql="delete from {} where _id = %(id)s".format(self.tablename)
        cursor=self.db.cursor()
        cursor.executemany(sql, ids)
        self.db.commit()
        cursor.close()

    def is_exist(self, id):
        sql="select count(_id) as c from {} where _id = %(id)s".format(
            self.tablename)
        cursor=self.db.cursor()
        cursor.execute(sql, {"id": id})
        rs=cursor.fetchone()
        cursor.close()              
        return rs[0] > 0

    def is_exist_list(self, ids):
        idstr = "','".join(ids)
        sql="select _id from {} where _id in ('{}')".format(
            self.tablename, idstr)        
        cursor = self.db.cursor()
        cursor.execute(sql)
        rsdata = cursor.fetchall()
        cursor.close()              
        rs = {}
        for item in rsdata:
            rs[item[0]] = 1        
        return rs

    def insert_or_update(self, datas):
        if len(datas) == 0:
            return
        data = datas[0]
        cols=", ".join([k for k in data])
        vals=", ".join(["%({})s".format(k) for k in data])
        set_arrays=[]
        for k in data:
            if k == "_id":
                continue
            set_arrays.append("{} = %({})s".format(k, k))
        set_strs=", ".join(set_arrays)
        sql = ""
        if self.type == "m":
            sql = "INSERT INTO {}({}) VALUES({}) ON DUPLICATE KEY UPDATE {};".format(self.tablename, cols, vals, set_strs)
        else:
            sql = """
            IF EXISTS (select _id from {} where _id = %(_id)s) 
		        BEGIN
			        UPDATE {} SET {} WHERE _id = %(_id)s
		        END
	        ELSE
		        BEGIN
			        INSERT INTO {}({}) VALUES({})
		        END
            """.format(self.tablename, self.tablename, set_strs, self.tablename, cols, vals)
        
        print(sql)
        cursor=self.db.cursor()
        cursor.executemany(sql, datas)
        self.db.commit()
        cursor.close()


    def insert(self, datas):
        if len(datas) == 0:
            return
        data = datas[0]
        cols=", ".join([k for k in data])
        vals=", ".join(["%({})s".format(k) for k in data])
        sql="INSERT INTO {}({}) VALUES({});".format(self.tablename, cols, vals)        
        cursor=self.db.cursor()
        cursor.executemany(sql, datas)
        self.db.commit()
        cursor.close()

    def update(self, datas):
        if len(datas) == 0:
            return
        data = datas[0]
        set_arrays=[]
        for k in data:
            if k == "_id":
                continue
            set_arrays.append("{} = %({})s".format(k, k))
        set_strs=", ".join(set_arrays)
        sql="UPDATE {} SET {} WHERE _id = %(_id)s".format(self.tablename, set_strs)

        cursor=self.db.cursor()
        cursor.executemany(sql, datas)
        self.db.commit()
        cursor.close()
    
    def append(self, id, op, data):
        
        idstr = str(id)               
        if op == "i" or op == "u":
            d = self.convert(data)
            for k in d:
                d[k] = self.get_val(d[k])
            self.other_list.append((idstr, d))
            self.other_ids.append(idstr)          
            self.count = self.count + 1
        elif op == "d":
            self.del_list.append({"id": idstr})
            self.count = self.count + 1
            
    def commit(self):
        add_dic = {}
        if(len(self.other_ids) > 0):
           other_datas = self.is_exist_list(self.other_ids)
           for item in self.other_list:

               if item[0] in other_datas or item[0] in add_dic:
                   self.up_list.append(item[1])
               else:
                   add_dic[item[0]] = 1
                   self.add_list.append(item[1])
        #data_list = [item[1] for item in self.other_list]
        
               
        add_len = len(self.add_list)
        up_len = len(self.up_list)
        # data_len = len(data_list)
        del_len = len(self.del_list)
        # print("i:{},u:{},d:{}".format(add_len, up_len, del_len))
        if add_len > 0:
            self.insert(self.add_list)            
            print("新增数据{}条".format(add_len))
        
        if up_len > 0:
           self.update(self.up_list)            
           print("更新数据{}条".format(up_len))

        if del_len > 0:
            self.del_data(self.del_list)            
            print("删除数据{}条".format(del_len))

        self.add_list = []
        self.up_list = []
        self.del_list = []
        self.other_ids = []
        self.other_list = []         
        self.count = 0       



class MongoSync:
    def __init__(self, conStr: str, sqldb: SqlSata):
        self.myclient = pymongo.MongoClient(conStr)
        self.local = self.myclient["local"]
        self.oplog = self.local["oplog.rs"]
        self.long_ts = 0
        self.db = sqldb
        self.now = datetime.datetime.now()
        self.count = 0
        self.proc_time = datetime.datetime.now()

    def get_last_ts(self):
        last_oplog = self.oplog.find().sort(
            "$natural", pymongo.DESCENDING).limit(-1).next()
        last_ts = bson_ts_to_long(last_oplog["ts"])
        return last_ts

    def get_last_run_ts(self):
        f = open("./mongosync.ts.config", "r")
        line = f.readline()                
        f.close()
        ts = int(line)
        return long_to_bson_ts(ts)
    def save_last_run_ts(self, ts):
        f = open("./mongosync.ts.config", "w")
        f.write(str(ts))
        f.close()

    def other_query(self, query):
        pass

    def get_op_datas(self, ts, skip, limit):
        query = {"op": {"$ne": "n"}}
        query["ts"] = {"$gte": ts}
        self.other_query(query)
        
        return self.oplog.find(query).sort("$natural", pymongo.ASCENDING).skip(skip).limit(limit)

    def full_run(self, ns: str):
        c = self.get_collection(ns)
        query = {}
        self.other_query(query)
        skip = 0
        limit = 1000
               
        while True:
            add_list = []
            update_list = []
            id_list = []
            all_data = []
            
            datas = c.find(query).skip(skip).limit(limit)
            skip = skip + limit
            for item in datas:
                idstr = str(item["_id"])
                d = self.db.convert(item)
                for k in d:
                    d[k] = self.db.get_val(d[k])
                all_data.append(d)
                id_list.append(idstr)
            if len(id_list) > 0:
                find_ids = self.db.is_exist_list(id_list)
                for item in all_data:
                    id = item["_id"]
                    if id in find_ids:
                        update_list.append(item)
                    else:
                        add_list.append(item)
                add_len = len(add_list)
                up_len = len(update_list)
                if add_len > 0:
                    self.db.insert(add_list)
                    print("新增数据{}条".format(add_len))
                if up_len > 0:
                    self.db.update(update_list)
                    print("更新数据{}条".format(up_len))
            else:
                break

            

    def run(self):
        ts = self.get_last_run_ts()    
        limit = 1000
        skip = 0
        while True:
            pre_ts = ts
            datas = self.get_op_datas(ts, skip, limit)   
            if (datetime.datetime.now() - self.now).seconds > 4:
                self.now = datetime.datetime.now()
                self.save_last_run_ts(self.long_ts)
                print("{} - 已处理{}条, ts {} ".format(self.proc_time, self.count, ts))
            pre_count = self.count
            for data in datas:
                self.count = self.count + 1
                op = data["op"]                
                if ts < data["ts"]:
                    ts = data["ts"]
                self.proc_time = data["wall"]
                self.long_ts = bson_ts_to_long(data["ts"])

                if not (op == "u" or op == "i" or op == "d"):
                    continue                  
                ns = data["ns"]
                #print(ns)
                if ns.startswith("system."):
                    continue
                if ns.startswith("config."):
                    continue
                
                id, d = self.get_data(op, data)
                #print(d)
                if not (op != "d" and (d == None or len(d) == 0)):
                    self.proc(ns, op, id, self.long_ts, d)
               
            datas_len = self.count - pre_count            
            
            if datas_len > 0 and ts == pre_ts:
                skip = skip + limit
                print("skip")           
            elif datas_len > 0:
                skip = 0
            else:
                break
         

    def get_data(self, op: str, data):
        if op == "i":
            rs = data["o"]
            return rs["_id"], rs
        elif op == "u":
            id = data["o2"]["_id"]
            rs = self.find_data(data["ns"], id)
            return id, rs
        elif op == "d":
            id = data["o"]["_id"]
            return id, {}
        return 0, {}

    def find_data(self, ns: str, id):
        # print(type(id).__name__)
        c = self.get_collection(ns)
        rs = c.find_one({"_id": id})
        return rs
    
    def find_data_list(self, ns: str, ids):
        c = self.get_collection(ns)
        rs = c.find({"_id": {"$in": ids }})
        return rs

    def get_collection(self, ns: str):
        database, coll = ns.split(".", 1)
        return self.myclient[database][coll]

    def proc(self, ns: str, op: str, id, ts: int, data):
        self.db.append(id, op, data)
        if self.db.count >= 1000:
            self.commit()
        #print("ns:{},op:{},id:{}, ts:{}".format(ns, op, id, ts))
    
    def commit(self):
        self.db.commit()
        self.save_last_run_ts(self.long_ts)

    def end(self):
        self.commit()  